/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import static org.sss.common.Utils.newInstance;
import static org.sss.server.ServerUtil.listDataReceiver;

import org.slf4j.Logger;
import org.sss.common.Bucket;
import org.sss.common.DataReceiver;
import org.sss.common.EncodingOutputQueue;
import org.sss.mapreduce.Broadcaster;
import org.sss.mapreduce.Encoder;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Partitioner;
import org.sss.mapreduce.SssException;
import org.sss.util.CallbackInterface;
import org.sss.util.Factory;

/**
 * Wrapper of EncodingOutputQueue
 */
public class TaskOutputImpl implements TaskOutput {
  private static Logger logger = SssServer.getLogger(TaskOutputImpl.class);
  private EncodingOutputQueue queue;
  private final GroupID groupID;
  private final int multiplicity;
  private final int queueLimitNBytes;
  private final Factory<Encoder> keyEncoderFactory;
  private final Factory<Encoder> valueEncoderFactory;
  private final Class<? extends Partitioner> partitionerClass;
  private final CallbackInterface<Throwable> onCompletion;
  private final SssServerResource sssResource;

  public TaskOutputImpl(GroupID outputGID,
      Factory<Encoder> keyEncoderFactory,
      Factory<Encoder> valueEncoderFactory,
      Class<? extends Partitioner> partitionerClass,
      long multiplicity,
      long queueLimitNBytes,
      SssServerResource sssResource,
      CallbackInterface<Throwable> onCompletion) throws SssException {
    this.multiplicity = (int)multiplicity;
    this.queueLimitNBytes = (int)queueLimitNBytes;
    this.keyEncoderFactory = keyEncoderFactory;
    this.valueEncoderFactory = valueEncoderFactory;
    if (outputGID.isBroadcast()) {
      if (partitionerClass != Broadcaster.class) {
        logger.warn("GroupID has broadcast flag. thus \"{}\" is not used.",
            partitionerClass.getName());
      }
      partitionerClass = Broadcaster.class;
    } else if (partitionerClass == Broadcaster.class) {
      throw new SssException("GroupID is not broadcast, but partitioner is Broadcaster.");
    }
    this.partitionerClass = partitionerClass;
    this.sssResource = sssResource;
    this.onCompletion = onCompletion;
    this.groupID = outputGID;
    renew();
  }

  @Override
  public void renew() throws SssException {
    this.queue = new EncodingOutputQueue(
        groupID, keyEncoderFactory, valueEncoderFactory,
        sssResource.getStorageNodeCount(),
        newInstance(partitionerClass), 
        sssResource.getIOTPWrite(),
        sssResource.getIOManager().createDataWriter(), 
        multiplicity, queueLimitNBytes,
        onCompletion);
  }

  @Override
  public DataReceiver<Bucket> getTLPutter(int limit) throws SssException {
    return new CacheDataReceiver<Bucket>(limit, listDataReceiver(queue));
  }

  @Override
  public void close() throws SssException {
    queue.close();
  }

  @Override
  public void setError(Throwable e) {
    queue.setError(e);
  }

  @Override
  public Throwable getError() {
    return queue.getError();
  }

  @Override
  public GroupID getGroupID() {
    return groupID;
  }

  public long[] getOutputCountCopy() {
    return queue.getOutputCountCopy();
  }

  public long[] getOutputNBytesCopy() {
    return queue.getOutputNBytesCopy();
  }
}
